package cn.gwm.flink.streaming.function.map;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName OriginalDataMapFunction
 * @Author xzh
 * @Date 2023/2/16 11:01
 **/
@Slf4j
public class OriginalDataMapFunction implements FlatMapFunction<String, JSONObject> {

    private String vehicleType;

    private Integer minLen;

    BeanSource.SourceEnum sourceEnum;

    //接收字段集合列表
    List<String> set;

    public OriginalDataMapFunction(String vehicleType, Integer minLen) {
        this.vehicleType = vehicleType;
        this.minLen = minLen;
    }

    public OriginalDataMapFunction(String vehicleType, Integer minLen, BeanSource.SourceEnum sourceEnum) {
        this.vehicleType = vehicleType;
        this.minLen = minLen;
        this.sourceEnum = sourceEnum;
    }

    public OriginalDataMapFunction(String vehicleType, Integer minLen, BeanSource.SourceEnum sourceEnum, List<String> set) {
        this.vehicleType = vehicleType;
        this.minLen = minLen;
        this.sourceEnum = sourceEnum;
        this.set = set;
    }

    @Override
    public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
        JSONObject object = null;
        try {
            object = JSONUtil.parseObj(s);
//            if("LGWFGUA68PM000032".equals(object.getStr("deviceId"))||"LGWFGUA62NB000004".equals(object.getStr("deviceId"))){
//                System.out.println("size="+object.size()+" model_code="+object.getStr("model_code")+" command="+object.getStr("command")+" deviceId="+object.getStr("deviceId")+" item_time="+object.getStr("item_time")+" data="+object.toString());
//            }
        } catch (Exception e) {
            log.error("源数据转json异常，源数据格式有误！");
            e.printStackTrace();
            throw e;
        }
        if (object.size() > minLen) {
            if (sourceEnum != null && (null==set||set.size()<1)) {
                try {
                    List<String> list = BeanSource.sourceList(sourceEnum);
                    JSONObject resultOjb = JSONUtil.createObj();
                    List<String> noFieldList = new ArrayList<>();
                    for (String feild : list) {
                        resultOjb.set(feild, object.getStr(feild));
                        if(!object.containsKey(feild)){
                            noFieldList.add(feild);
                        }
                    }
                    collectData(collector, resultOjb);
                    log.info("{}车型数据中，没有找到配置字段{}",vehicleType, JSON.toJSONString(noFieldList));
                } catch (Exception e) {
                    log.error("缩减字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
                    e.printStackTrace();
                    throw e;
                }
            }else if(set!=null&&set.size()>0){
                try {
                    JSONObject resultOjb = JSONUtil.createObj();
                    List<String> noFieldList = new ArrayList<>();
                    for (String feild : set) {
                        resultOjb.set(feild, object.getStr(feild));
                        if(!object.containsKey(feild)){
                            noFieldList.add(feild);
                        }
                    }
                    collectData(collector, resultOjb);
                    log.info("{}车型数据中，没有找到配置字段{}",vehicleType, JSON.toJSONString(noFieldList));
                } catch (Exception e) {
                    log.error("缩减字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
                    e.printStackTrace();
                    throw e;
                }
            } else {
                try {
                    //gps 过滤有效数据，仅保留经纬度正常数据
                    if (object.containsKey(BaseFields.command) && "TBOX_MONITOR".equals(object.getStr(BaseFields.command))) {
                        if(object.containsKey("tboxMoniGPSstatus")&&object.containsKey("tboxMoniLongitude")&&object.containsKey("tboxMoniLatitude")){
                            //过滤仅中国境内的gps信号
                            if (object.getInt("tboxMoniGPSstatus") == 0 && object.getDouble("tboxMoniLongitude") > 72
                                    && object.getDouble("tboxMoniLongitude") < 136 && object.getDouble("tboxMoniLatitude") > 3
                                    && object.getDouble("tboxMoniLatitude") < 54) {
                                collectData(collector, object);
                            }
                        }else{
                            log.error("经纬度过滤字段数据异常！object={}", object.toString());
                        }
                    } else {
                        collectData(collector, object);
                    }
                } catch (Exception e) {
                    log.error("经纬度过滤字段数据异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
                    e.printStackTrace();
                    throw e;
                }
            }
        }
    }

    private void collectData(Collector<JSONObject> collector, JSONObject object) {
        try {
            object.set(BaseFields.STORE_TIME, System.currentTimeMillis())
                    .set(VehicleType.VEHICLE_TYPE, vehicleType)
                    .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                            DatePattern.NORM_DATE_PATTERN));
        } catch (Exception e) {
            log.error("json字段不存在item_time字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
            e.printStackTrace();
            throw e;
        }
        collector.collect(object);

    }
}
